#include <stdio.h>
#include "mysql_connection.h"
#include "SQLcommand_impl.h"
#include "database_module_impl.h"
#include "mysql_query_result.h"

#ifndef WIN32
#define Sleep usleep
#endif

#pragma comment(lib, "libmysql.lib")

namespace utility
{
namespace database
{
    mysql_connection::mysql_connection(uint32 index, database_module_impl * db_module): 
        _index(index), _mysql_object(NULL), _connected(false), _db_module(db_module), 
        _worker_thread(NULL), _terminated(true)
    {
        db_module->register_runnable_object(this);
    }

    mysql_connection::~mysql_connection()
    {
        _db_module->unregister_runnable_object(this);
        disconnect();
        
    }

    bool mysql_connection::connect(const char * host, const char * user, const char * password, const char * db_name, uint32 port)
    {
        disconnect();

        if (_mysql_object == NULL)
            _mysql_object = mysql_init(NULL);
        if (_mysql_object == NULL)
            return false;
        _connected = mysql_real_connect(_mysql_object, 
            host, user, password, 
            db_name, port, NULL, 
            CLIENT_MULTI_STATEMENTS) != NULL;
#ifdef _DEBUG
        if (!_connected)
            printf("connection failed : %s", mysql_error(_mysql_object));
#endif
        if (_connected)
        {
            _terminated = false;
            _worker_thread = new boost::thread(boost::bind(&mysql_connection::_start_work, this));
        }
        return _connected;
    }

    void mysql_connection::disconnect()
    {
        _terminated = true;
        if (_worker_thread)
        {
            _worker_thread->join();
            delete _worker_thread;
            _worker_thread = NULL;
        }

        run();

        if (_mysql_object)
        {
            mysql_close(_mysql_object);
            _mysql_object = NULL;
        }

        _connected = false;
    }

    SQLcommand_pointer mysql_connection::create_command(char * SQLtext)
    {
        return SQLcommand_pointer(new SQLcommand_impl(SQLtext, this));
    }
        
    void mysql_connection::async_execute(SQLcommand_pointer command)
    {
        //boost::lock_guard<lock_type> guard(_lock_waiting);
        //_waiting_commands.push_back(command);
        _waiting_commands.offer(command);
    }

    uint32 mysql_connection::escape_string(const char * src, size_t src_length, char * dst)
    {
        try
        {
            return mysql_real_escape_string(_mysql_object, dst, src, src_length);
        }
        catch (...)
        {
            throw error_code(3000, "error occurred when escaping string");
            return 0;
        }
    }

    bool mysql_connection::run()
    {
        bool busy = false;
        while (_completed_commands.size() > 0)
        {
            const completed_item_pointer item = _completed_commands.poll();
            (item->first)->get_callback_function()->query_callback(
                item->second._result, 
                item->second._error
                );
            busy = true;
        }
        return busy;
    }

    void mysql_connection::_start_work()
    {
        while(!_terminated || _waiting_commands.size() > 0)
        {
            int busy = false;
            while(_waiting_commands.size() > 0)
            {
                SQLcommand_pointer command = _waiting_commands.poll(); 
                if (SQLcommand::QUERY == command->get_type())
                {
                    _execute_query(command);
                }
                else if (SQLcommand::NON_QUERY == command->get_type())
                {
                    _execute_non_query(command);
                }
                else if (SQLcommand::CALL_PROCEDURE == command->get_type())
                {
                    _call_procedure(command);
                }
                _clear_mysql_res_cache();
                busy = true;
            }
            if (!busy)
                Sleep(1);

        }
        //printf("exit thread\n");
    }

    void mysql_connection::_execute_query(SQLcommand_pointer command)
    {
        completed_item_pointer item = completed_item_pointer(new completed_item);
        item->first = command;

        try
        {
            std::string SQL = command->get_SQL();
            int rst = mysql_real_query(_mysql_object, SQL.c_str(), SQL.length());
            if (rst != 0)
            {
                item->second._error.set_error(rst, mysql_error(_mysql_object));
            }
            else
            {
                MYSQL_RES * res = mysql_store_result(_mysql_object);
                if (res == NULL)
                {
                    item->second._error.set_error(mysql_errno(_mysql_object), mysql_error(_mysql_object));
                }
                else
                {
                    item->second._result = new mysql_query_result(res);
                    mysql_free_result(res);
                }
            }
        }
        catch(error_code & ec)
        {
            item->second._error = ec;
        }

        //_completed_commands.push_back(item);
        _completed_commands.offer(item);
    }

    void mysql_connection::_execute_non_query(SQLcommand_pointer command)
    {
        completed_item_pointer item = completed_item_pointer(new completed_item);
        item->first = command;

        try
        {
            std::string SQL = command->get_SQL();
            if (mysql_real_query(_mysql_object, SQL.c_str(), SQL.length()) != 0)
            {
                item->second._error.set_error(mysql_errno(_mysql_object), mysql_error(_mysql_object));
            }
            else
            {
                _clear_mysql_res_cache();
            }
        }
        catch(error_code & ec)
        {
            item->second._error = ec;
        }

        //_completed_commands.push_back(item);
        _completed_commands.offer(item);
    }

    void mysql_connection::_call_procedure(SQLcommand_pointer command)
    {
        completed_item_pointer item = completed_item_pointer(new completed_item);
        item->first = command;

        try
        {
            std::string SQL = command->get_SQL();
            int rst = mysql_real_query(_mysql_object, SQL.c_str(), SQL.length());
            if (rst != 0)
            {
                item->second._error.set_error(mysql_errno(_mysql_object), mysql_error(_mysql_object));
            }
            else
            {
                _clear_mysql_res_cache();
                SQL = command->get_procedure_output_value();
                if (SQL.length() > 0)
                {
                    int rst = mysql_real_query(_mysql_object, SQL.c_str(), SQL.length());
                    if (rst != 0)
                    {
                        item->second._error.set_error(mysql_errno(_mysql_object), mysql_error(_mysql_object));
                    }
                    else
                    {
                        MYSQL_RES * res = mysql_store_result(_mysql_object);
                        if (res == NULL)
                        {
                            item->second._error.set_error(mysql_errno(_mysql_object), mysql_error(_mysql_object));
                        }
                        else
                        {
                            item->second._result = new mysql_query_result(res);
                            mysql_free_result(res);
                        }
                    }
                }
            }
        }
        catch(error_code & ec)
        {
            item->second._error = ec;
        }

        //_completed_commands.push_back(item);
        _completed_commands.offer(item);
    }

    void mysql_connection::_clear_mysql_res_cache()
    {
        MYSQL_RES * my_res = mysql_store_result(_mysql_object);
        if (my_res)
            mysql_free_result(my_res);
        while(mysql_next_result(_mysql_object) == 0)
        {
            mysql_free_result(mysql_store_result(_mysql_object));
        }
    }
}
}